package com.gph.saviorframework.mq.api.impl.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.protocol.ResponseCode;
import com.alibaba.rocketmq.remoting.exception.RemotingConnectException;
import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
import com.gph.saviorframework.mq.api.Message;
import com.gph.saviorframework.mq.api.Producer;
import com.gph.saviorframework.mq.api.PropertyKeyConst;
import com.gph.saviorframework.mq.api.SendResult;
import com.gph.saviorframework.mq.api.exception.MQClientException;
import com.gph.saviorframework.mq.api.impl.util.MQUtil;
import com.gph.saviorframework.mq.api.impl.util.NameAddrUtils;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.alibaba.rocketmq.common.UtilAll.getPid;

/**
 * 生产者实现
 * @author guopeihui
 */
public class ProducerImpl extends MQClientAbstract implements Producer {

    private final DefaultMQProducer defaultMQProducer;

    private final AtomicBoolean started = new AtomicBoolean(false);


    public ProducerImpl(final Properties properties){

        super(properties);

        this.defaultMQProducer = new DefaultMQProducer(new ClientRPCHook(this.sessionCredentials));

        String producerGroup = properties.getProperty(PropertyKeyConst.ProducerId, "__GPH_PRODUCER_DEFAULT_GROUP");
        this.defaultMQProducer.setProducerGroup(producerGroup);

        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        this.defaultMQProducer.setVipChannelEnabled(isVipChannelEnabled);

        if (properties.containsKey(PropertyKeyConst.SendMsgTimeoutMillis)) {
            this.defaultMQProducer.setSendMsgTimeout(Integer.valueOf(properties.get(PropertyKeyConst.SendMsgTimeoutMillis).toString()));
        }
        else {
            this.defaultMQProducer.setSendMsgTimeout(5000);
        }

        nameServerAddr = properties.getProperty(PropertyKeyConst.NAMESRV_ADDR);
        this.defaultMQProducer.setNamesrvAddr(nameServerAddr);

        this.defaultMQProducer.setInstanceName(buildIntanceName());

        /**
         * 消息最大是4M
         */
        this.defaultMQProducer.setMaxMessageSize(1024 * 1024 * 4);
    }


    /**
     * 启动服务
     */
    public void start() {
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQProducer.start();
            }
        } catch (Exception e) {
            throw new MQClientException(e.getMessage());
        }
    }

    /**
     * 关闭服务
     */
    public void shutdown() {
        if (this.started.compareAndSet(true, false)) {
            this.defaultMQProducer.shutdown();
        }
    }

    /**
     * 同步发送消息，只要不抛异常就表示成功
     *
     * @param message
     * @return 发送结果，含消息Id
     */
    public SendResult send(Message message) {
        this.checkProducerServiceState(this.defaultMQProducer.getDefaultMQProducerImpl());
        com.alibaba.rocketmq.common.message.Message msgRMQ = MQUtil.msgConvert(message);
        try {
            com.alibaba.rocketmq.client.producer.SendResult sendResultRMQ = this.defaultMQProducer.send(msgRMQ);

            SendResult sendResult = new SendResult();
            sendResult.setMessageId(sendResultRMQ.getMsgId());
            message.setMsgID(sendResultRMQ.getMsgId());
            return sendResult;
        }
        catch (Exception e) {
            this.checkProducerException(e, message);
            return null;
        }
    }

    /**
     * 发送消息，Oneway形式，服务器不应答，无法保证消息是否成功到达服务器
     *
     * @param message
     */
    public void sendOneway(Message message) {
        this.checkProducerServiceState(this.defaultMQProducer.getDefaultMQProducerImpl());

        com.alibaba.rocketmq.common.message.Message msgRMQ = MQUtil.msgConvert(message);
        try {
            this.defaultMQProducer.sendOneway(msgRMQ);
        }
        catch (Exception e) {
            this.checkProducerException(e, message);
        }
    }

    public boolean isStarted() {
        return started.get();
    }

    public boolean isClosed() {
        return !isStarted();
    }

    private void checkProducerException(Exception e, Message message) {
        if (e instanceof com.alibaba.rocketmq.client.exception.MQClientException) {
            //
            if (e.getCause() != null) {
                // 无法连接Broker
                if (e.getCause() instanceof RemotingConnectException) {
                    throw new MQClientException(String.format("Connect broker failed, Topic: %s", message.getTopic()));
                }
                // 发送消息超时
                else if (e.getCause() instanceof RemotingTimeoutException) {
                    throw new MQClientException(String.format("Send message to broker timeout, %dms, Topic: %s",
                            this.defaultMQProducer.getSendMsgTimeout(), message.getTopic()));
                }
                // Broker返回异常
                else if (e.getCause() instanceof MQBrokerException) {
                    MQBrokerException excep = (MQBrokerException) e.getCause();
                    throw new MQClientException(String.format("Receive a broker exception, Topic: %s, %s", message.getTopic(), excep.getErrorMessage()));
                }
            }
            // 纯客户端异常
            else {
                com.alibaba.rocketmq.client.exception.MQClientException excep = (com.alibaba.rocketmq.client.exception.MQClientException) e;
                if (-1 == excep.getResponseCode()) {
                    throw new MQClientException(String.format("Topic does not exist, Topic: %s", message.getTopic()));
                }
                else if (ResponseCode.MESSAGE_ILLEGAL == excep.getResponseCode()) {
                    throw new MQClientException(String.format("MQ Client check message exception, Topic: %s", message.getTopic()));
                }
            }
        }

        throw new MQClientException("defaultMQProducer send exception", e);
    }
}
